package com.example.consumer;

import com.example.constant.RedisConstant;
import com.example.util.redis.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.stream.Collectors;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/5/31 20:27
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Component
@Slf4j
public class DeliveryRunner {
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 定时任务。处理消息转移过来的数据
     */
    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 60 * 5)
    @Async(value = "forkJoinPool")
    public void dealWithDelivery() {
        StreamOperations<String, String, String> streamOperations = RedisUtil.getStringStringStringStreamOperations(this.redisTemplate);
        // 获取消费者组中的pending消息信息，本质上就是执行XPENDING指令
        PendingMessagesSummary pendingMessagesSummary = streamOperations.pending(RedisConstant.COMMENT_KEY,RedisConstant.COMMENT_GROUP);

        showPendingInfo(pendingMessagesSummary);

        processPendingQuene(RedisConstant.COMMENT_KEY,RedisConstant.COMMENT_GROUP,pendingMessagesSummary,streamOperations);

    }

    private void processPendingQuene(String streamKey,String groupName,
                                     PendingMessagesSummary pendingMessagesSummary,
                                     StreamOperations<String, String, String> streamOperations) {

        Vector<MapRecord> deliveryList = new Vector<>();
        pendingMessagesSummary.getPendingMessagesPerConsumer().forEach((consumerName,count) -> {
            Consumer consumer = Consumer.from(groupName,consumerName);
            PendingMessages pendingMessages = streamOperations.pending(streamKey,consumer);
            Vector<MapRecord> recordVector = pendingMessages
                    .stream()
                    .parallel()
                    .filter(pendingMessage -> {
                        // 只消费转组次数大于1次的
                        return pendingMessage.getTotalDeliveryCount() > 1;
                    }).map(pendingMessage -> {
                        showMessageInfo(pendingMessage);
                        // 通过streamOperations，直接读取这条pending消息，
                        RecordId id = pendingMessage.getId();
                        List<MapRecord<String, String, String>> result = streamOperations.range(streamKey,Range.rightOpen(id.toString(),id.toString()));
                        // 开始和结束都是同一个ID，所以结果只有一条
                        MapRecord<String, String, String> entries = result.get(0);
                        return entries;
                    }).collect(Collectors.toCollection(Vector::new));
            if (!CollectionUtils.isEmpty(recordVector)) {
                deliveryList.addAll(recordVector);
                log.warn(" 消息转移 recordList:{}",recordVector.size());
            }
        });

        if (!CollectionUtils.isEmpty(deliveryList)) {
            try {
                log.info("deliveryList  size:{}, deliveryList:{}",deliveryList.size(),deliveryList);
                List<RecordId> collect = deliveryList.stream().map(MapRecord::getId).collect(Collectors.toList());
                Thread.sleep(2000);
                RecordId[] recordIds = collect.toArray(new RecordId[collect.size()]);
                Long acknowledge = streamOperations.acknowledge(RedisConstant.COMMENT_KEY,groupName,recordIds);
                log.warn("手动ack 消息转移过来的数据acknowledge count:{}",acknowledge);
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error("liyuhang: 捕捉到异常 {} , 异常栈信息:",e.getMessage(),e);
            }
        }

    }

    /**
     * 显示消息
     * @param message
     */
    private void showMessageInfo(PendingMessage message) {
        // 消息的ID
        RecordId recordId = message.getId();
        // 消息从消费组中获取，到此刻的时间
        Duration elapsedTimeSinceLastDelivery = message.getElapsedTimeSinceLastDelivery();
        // 消息被获取的次数
        long deliveryCount = message.getTotalDeliveryCount();
        log.info("pending 消息，id={}, elapsedTimeSinceLastDelivery={}, deliveryCount={}",recordId,elapsedTimeSinceLastDelivery,deliveryCount);
    }

    /**
     * 消息pending总揽
     * @param pendingMessagesSummary
     */
    private void showPendingInfo(PendingMessagesSummary pendingMessagesSummary) {
        // 所有pending消息的数量
        long totalPendingMessages = pendingMessagesSummary.getTotalPendingMessages();
        String groupName = pendingMessagesSummary.getGroupName();
        // pending队列中的最小ID
        String minMessageId = pendingMessagesSummary.minMessageId();
        // pending队列中的最大ID
        String maxMessageId = pendingMessagesSummary.maxMessageId();
        // 每个消费者的pending消息数量
        Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
        log.info("消费组：{}，一共有{}条pending消息，最大ID={}，最小ID={},各个消费组情况:{}",groupName,totalPendingMessages,minMessageId,maxMessageId,pendingMessagesPerConsumer);
        log.warn("pendingMessagesSummary:{}",pendingMessagesSummary);
    }
}
